#![allow(clippy::disallowed_types)]
/// This module contains the route handlers for the PocketIc server.
///
/// A handler may receive a representation of a PocketIc Operation in the request
/// body. This has to be canonicalized into a PocketIc Operation before we can
/// deterministically update the PocketIc state machine.
///
use super::state::{
    ApiState, DEFAULT_SYNC_WAIT_DURATION, OpOut, PocketIcError, StateLabel, UpdateReply,
};
use crate::pocket_ic::{
    AddCycles, AwaitIngressMessage, CallRequest, CallRequestVersion, CanisterReadStateRequest,
    CanisterSnapshotDownload, CanisterSnapshotUpload, DashboardRequest, GetCanisterHttp,
    GetControllers, GetCyclesBalance, GetStableMemory, GetSubnet, GetTime, GetTopology,
    IngressMessageStatus, MockCanisterHttp, PubKey, Query, QueryRequest, SetCertifiedTime,
    SetStableMemory, SetTime, StatusRequest, SubmitIngressMessage, SubnetReadStateRequest, Tick,
};
use crate::{BlobStore, InstanceId, OpId, Operation, async_trait, pocket_ic::PocketIc};
use aide::{
    NoApi,
    axum::ApiRouter,
    axum::routing::{ApiMethodRouter, delete, get, post},
};

use axum::{
    Json,
    body::{Body, Bytes},
    extract::{self, Path, Request, State},
    http::{self, HeaderMap, HeaderName, StatusCode},
    middleware::Next,
    response::{IntoResponse, Response},
};
use axum_extra::headers;
use axum_extra::headers::HeaderMapExt;
use backoff::backoff::Backoff;
use backoff::{ExponentialBackoff, ExponentialBackoffBuilder};
use ic_boundary::{ErrorClientFacing, MAX_REQUEST_BODY_SIZE};
use ic_http_endpoints_public::{cors_layer, make_plaintext_response, query, read_state};
use ic_registry_routing_table::RoutingTable;
use ic_types::{CanisterId, SnapshotId, SubnetId};
use pocket_ic::RejectResponse;
use pocket_ic::common::rest::{
    self, ApiResponse, AutoProgressConfig, ExtendedSubnetConfigSet, HttpGatewayConfig,
    HttpGatewayDetails, IcpConfig, IcpFeatures, InitialTime, InstanceConfig,
    MockCanisterHttpResponse, RawAddCycles, RawCanisterCall, RawCanisterHttpRequest, RawCanisterId,
    RawCanisterResult, RawCanisterSnapshotDownload, RawCanisterSnapshotId,
    RawCanisterSnapshotUpload, RawCycles, RawIngressStatusArgs, RawMessageId,
    RawMockCanisterHttpResponse, RawPrincipalId, RawSetStableMemory, RawStableMemory, RawSubnetId,
    RawTime, TickConfigs, Topology,
};
use serde::Serialize;
use slog::Level;
use std::str::FromStr;
use std::{
    collections::BTreeMap,
    fs::File,
    sync::Arc,
    sync::atomic::AtomicU64,
    time::{Duration, SystemTime},
};
use tokio::runtime::Runtime;
use tower_http::limit::RequestBodyLimitLayer;
use tracing::trace;

type PocketHttpResponse = (BTreeMap<String, Vec<u8>>, Vec<u8>);

pub(crate) async fn into_api_response(
    resp: Response,
) -> (StatusCode, BTreeMap<String, Vec<u8>>, Vec<u8>) {
    (
        resp.status(),
        resp.headers()
            .iter()
            .map(|(name, value)| (name.as_str().to_string(), value.as_bytes().to_vec()))
            .collect(),
        axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap()
            .to_vec(),
    )
}

/// Name of a header that allows clients to specify for how long their are willing to wait for a
/// response on a open http request.
pub static TIMEOUT_HEADER_NAME: HeaderName = HeaderName::from_static("processing-timeout-ms");

const RETRY_TIMEOUT_S: u64 = 300;

#[derive(Clone)]
pub struct AppState {
    pub api_state: Arc<ApiState>,
    pub pending_requests: Arc<AtomicU64>,
    /// TTL in nanoseconds since UNIX epoch
    pub min_alive_until: Arc<AtomicU64>,
    pub runtime: Arc<Runtime>,
    pub mainnet_routing_table: RoutingTable,
    pub blob_store: Arc<dyn BlobStore>,
}

pub fn instance_read_routes<S>() -> ApiRouter<S>
where
    S: Clone + Send + Sync + 'static,
    AppState: extract::FromRef<S>,
{
    ApiRouter::new()
        .directory_route("/query", post(handler_json_query))
        .directory_route("/topology", get(handler_topology))
        .directory_route("/get_time", get(handler_get_time))
        .directory_route("/get_canister_http", get(handler_get_canister_http))
        .directory_route("/get_controllers", post(handler_get_controllers))
        .directory_route("/get_cycles", post(handler_get_cycles))
        .directory_route("/get_stable_memory", post(handler_get_stable_memory))
        .directory_route("/get_subnet", post(handler_get_subnet))
        .directory_route("/pub_key", post(handler_pub_key))
        .directory_route("/ingress_status", post(handler_ingress_status))
}

pub fn instance_update_routes<S>() -> ApiRouter<S>
where
    S: Clone + Send + Sync + 'static,
    AppState: extract::FromRef<S>,
{
    ApiRouter::new()
        .directory_route(
            "/submit_ingress_message",
            post(handler_submit_ingress_message),
        )
        .directory_route(
            "/await_ingress_message",
            post(handler_await_ingress_message),
        )
        .directory_route("/set_time", post(handler_set_time))
        .directory_route("/set_certified_time", post(handler_set_certified_time))
        .directory_route("/add_cycles", post(handler_add_cycles))
        .directory_route("/set_stable_memory", post(handler_set_stable_memory))
        .directory_route("/tick", post(handler_tick))
        .directory_route("/mock_canister_http", post(handler_mock_canister_http))
        .directory_route(
            "/canister_snapshot_download",
            post(handler_canister_snapshot_download),
        )
        .directory_route(
            "/canister_snapshot_upload",
            post(handler_canister_snapshot_upload),
        )
}

async fn handle_limit_error(req: Request, next: Next) -> Response {
    let resp = next.run(req).await;
    if resp.status() == StatusCode::PAYLOAD_TOO_LARGE {
        return ErrorClientFacing::PayloadTooLarge(MAX_REQUEST_BODY_SIZE).into_response();
    }
    resp
}

pub fn instance_api_v2_routes<S>() -> ApiRouter<S>
where
    S: Clone + Send + Sync + 'static,
    AppState: extract::FromRef<S>,
{
    ApiRouter::new()
        .directory_route("/status", get(handler_status))
        .directory_route(
            "/canister/{ecid}/call",
            post(handler_call_v2)
                .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_SIZE))
                .layer(axum::middleware::from_fn(handle_limit_error)),
        )
        .directory_route(
            "/canister/{ecid}/query",
            post(handler_query_v2)
                .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_SIZE))
                .layer(axum::middleware::from_fn(handle_limit_error)),
        )
        .directory_route(
            "/canister/{ecid}/read_state",
            post(handler_canister_read_state_v2)
                .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_SIZE))
                .layer(axum::middleware::from_fn(handle_limit_error)),
        )
        .directory_route(
            "/subnet/{sid}/read_state",
            post(handler_subnet_read_state_v2)
                .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_SIZE))
                .layer(axum::middleware::from_fn(handle_limit_error)),
        )
}

pub fn instance_api_v3_routes<S>() -> ApiRouter<S>
where
    S: Clone + Send + Sync + 'static,
    AppState: extract::FromRef<S>,
{
    ApiRouter::new()
        .directory_route(
            "/canister/{ecid}/call",
            post(handler_call_v3)
                .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_SIZE))
                .layer(axum::middleware::from_fn(handle_limit_error)),
        )
        .directory_route(
            "/canister/{ecid}/query",
            post(handler_query_v3)
                .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_SIZE))
                .layer(axum::middleware::from_fn(handle_limit_error)),
        )
        .directory_route(
            "/canister/{ecid}/read_state",
            post(handler_canister_read_state_v3)
                .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_SIZE))
                .layer(axum::middleware::from_fn(handle_limit_error)),
        )
        .directory_route(
            "/subnet/{sid}/read_state",
            post(handler_subnet_read_state_v3)
                .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_SIZE))
                .layer(axum::middleware::from_fn(handle_limit_error)),
        )
}

pub fn instance_api_v4_routes<S>() -> ApiRouter<S>
where
    S: Clone + Send + Sync + 'static,
    AppState: extract::FromRef<S>,
{
    ApiRouter::new().directory_route(
        "/canister/{ecid}/call",
        post(handler_call_v4)
            .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_SIZE))
            .layer(axum::middleware::from_fn(handle_limit_error)),
    )
}

pub fn instances_routes<S>() -> ApiRouter<S>
where
    S: Clone + Send + Sync + 'static,
    AppState: extract::FromRef<S>,
{
    ApiRouter::new()
        //
        // List all IC instances.
        .api_route("/", get(list_instances))
        //
        // Create a new IC instance. Takes a SubnetConfig which must contain at least one subnet.
        // Returns an InstanceId.
        .api_route("/", post(create_instance))
        //
        // Deletes an instance.
        .directory_route("/{id}", delete(delete_instance))
        //
        // All the read-only endpoints
        .nest("/{id}/read", instance_read_routes())
        //
        // All the state-changing endpoints
        .nest("/{id}/update", instance_update_routes())
        //
        // All the api v2 endpoints
        .nest("/{id}/api/v2", instance_api_v2_routes())
        //
        // All the api v3 endpoints
        .nest("/{id}/api/v3", instance_api_v3_routes())
        //
        // All the api v4 endpoints
        .nest("/{id}/api/v4", instance_api_v4_routes())
        //
        // The instance dashboard
        .api_route("/{id}/_/dashboard", get(handler_dashboard))
        // The topology to be retrieved via an HTTP gateway that cannot route `/read/topology`.
        .api_route("/{id}/_/topology", get(handler_topology))
        // Configures an IC instance to make progress automatically,
        // i.e., periodically update the time of the IC instance
        // to the real time and execute rounds on the subnets.
        .api_route("/{id}/auto_progress", post(auto_progress))
        // Returns whether automatic progress is enabled for an IC instance.
        .api_route("/{id}/auto_progress", get(get_auto_progress))
        //
        // Stop automatic progress (see endpoint `auto_progress`)
        // on an IC instance.
        .api_route("/{id}/stop_progress", post(stop_progress))
        .layer(cors_layer())
}

pub fn http_gateway_routes<S>() -> ApiRouter<S>
where
    S: Clone + Send + Sync + 'static,
    AppState: extract::FromRef<S>,
{
    ApiRouter::new()
        // List all HTTP gateways.
        .api_route("/", get(list_http_gateways))
        // Create a new HTTP gateway instance. Takes a HttpGatewayConfig.
        // Returns an InstanceId and the HTTP gateway's port.
        .api_route("/", post(create_http_gateway))
        // Stops an HTTP gateway.
        .api_route("/{id}/stop", post(stop_http_gateway))
}

async fn run_operation<T: Serialize + FromOpOut>(
    api_state: Arc<ApiState>,
    instance_id: InstanceId,
    timeout: Option<Duration>,
    op: impl Operation + Send + Sync + 'static,
) -> (StatusCode, ApiResponse<T>) {
    let retry_if_busy = op.retry_if_busy();
    let op = Arc::new(op);
    let mut retry_policy: ExponentialBackoff = ExponentialBackoffBuilder::new()
        .with_initial_interval(Duration::from_millis(10))
        .with_max_interval(Duration::from_secs(1))
        .with_multiplier(2.0)
        .with_max_elapsed_time(Some(Duration::from_secs(RETRY_TIMEOUT_S)))
        .build();
    loop {
        match api_state
            .update_with_timeout(op.clone(), instance_id, timeout)
            .await
        {
            Err(e) => {
                break (
                    StatusCode::BAD_REQUEST,
                    ApiResponse::Error {
                        message: format!("{e:?}"),
                    },
                );
            }
            Ok(update_reply) => {
                match update_reply {
                    // If the op_id of the ongoing operation is the requested one, we return code 202.
                    UpdateReply::Started { state_label, op_id } => {
                        break (
                            StatusCode::ACCEPTED,
                            ApiResponse::Started {
                                state_label: base64::encode_config(state_label.0, base64::URL_SAFE),
                                op_id: op_id.0.to_string(),
                            },
                        );
                    }
                    // Otherwise, the instance is busy with a different computation, so we retry (if appliacable) or return 409.
                    UpdateReply::Busy { state_label, op_id } => {
                        if retry_if_busy {
                            trace!(
                                "run_operation::retry_busy instance_id={} state_label={:?} op_id={}",
                                instance_id, state_label, op_id.0
                            );
                            match retry_policy.next_backoff() {
                                Some(duration) => tokio::time::sleep(duration).await,
                                None => {
                                    break (
                                        StatusCode::TOO_MANY_REQUESTS,
                                        ApiResponse::Error {
                                            message: "Service is overloaded, try again later."
                                                .to_string(),
                                        },
                                    );
                                }
                            }
                        } else {
                            break (
                                StatusCode::CONFLICT,
                                ApiResponse::Busy {
                                    state_label: base64::encode_config(
                                        state_label.0,
                                        base64::URL_SAFE,
                                    ),
                                    op_id: op_id.0.to_string(),
                                },
                            );
                        }
                    }
                    UpdateReply::Output(op_out) => break FromOpOut::from(op_out).await,
                }
            }
        }
    }
}

#[derive(Copy, Clone, Debug)]
pub struct OpConversionError;

#[async_trait]
trait FromOpOut: Sized {
    async fn from(value: OpOut) -> (StatusCode, ApiResponse<Self>);
}

#[async_trait]
impl<T: TryFrom<OpOut>> FromOpOut for T {
    async fn from(value: OpOut) -> (StatusCode, ApiResponse<T>) {
        // match errors explicitly to make sure they have a 4xx status code
        match value {
            OpOut::Error(PocketIcError::Forbidden(msg)) => (
                StatusCode::FORBIDDEN,
                ApiResponse::Error {
                    message: msg.to_string(),
                },
            ),
            OpOut::Error(e) => (
                StatusCode::BAD_REQUEST,
                ApiResponse::Error {
                    message: format!("{e:?}"),
                },
            ),
            val => match T::try_from(val) {
                Ok(t) => (StatusCode::OK, ApiResponse::Success(t)),
                _ => (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    ApiResponse::Error {
                        message: "operation returned invalid type".into(),
                    },
                ),
            },
        }
    }
}

impl TryFrom<OpOut> for RawTime {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::Time(time) => Ok(RawTime {
                nanos_since_epoch: time,
            }),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for Topology {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::Topology(topology) => Ok(topology),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for () {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::NoOutput => Ok(()),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for Vec<RawPrincipalId> {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::Controllers(controllers) => Ok(controllers
                .into_iter()
                .map(|principal_id| principal_id.0.into())
                .collect()),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for RawCycles {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::Cycles(cycles) => Ok(RawCycles { cycles }),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for RawStableMemory {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::StableMemBytes(stable_memory) => Ok(RawStableMemory {
                blob: stable_memory,
            }),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for RawCanisterResult {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::CanisterResult(result) => Ok(result.into()),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for Option<RawCanisterResult> {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::CanisterResult(result) => Ok(Some(result.into())),
            OpOut::NoOutput => Ok(None),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for PocketIcError {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::Error(e) => Ok(e),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for RawCanisterId {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::CanisterId(canister_id) => Ok(RawCanisterId {
                canister_id: canister_id.get().to_vec(),
            }),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for Option<RawSubnetId> {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::MaybeSubnetId(maybe_subnet_id) => {
                Ok(maybe_subnet_id.map(|subnet_id| RawSubnetId {
                    subnet_id: subnet_id.get().to_vec(),
                }))
            }
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for Vec<u8> {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::Bytes(bytes) => Ok(bytes),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for Result<RawMessageId, RejectResponse> {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::MessageId((effective_principal, message_id)) => Ok(Ok(RawMessageId {
                effective_principal: effective_principal.into(),
                message_id,
            })),
            OpOut::CanisterResult(Err(reject_response)) => Ok(Err(reject_response)),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for Vec<RawCanisterHttpRequest> {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::CanisterHttp(canister_http_requests) => Ok(canister_http_requests
                .into_iter()
                .map(|r| r.into())
                .collect()),
            _ => Err(OpConversionError),
        }
    }
}

impl TryFrom<OpOut> for RawCanisterSnapshotId {
    type Error = OpConversionError;
    fn try_from(value: OpOut) -> Result<Self, Self::Error> {
        match value {
            OpOut::CanisterSnapshotId(snapshot_id) => Ok(RawCanisterSnapshotId { snapshot_id }),
            _ => Err(OpConversionError),
        }
    }
}

#[async_trait]
impl FromOpOut for PocketHttpResponse {
    async fn from(value: OpOut) -> (StatusCode, ApiResponse<PocketHttpResponse>) {
        async fn from_bn_err(
            err: ErrorClientFacing,
        ) -> (StatusCode, ApiResponse<PocketHttpResponse>) {
            let resp = err.into_response();
            let (status, headers, bytes) = into_api_response(resp).await;
            (status, ApiResponse::Success((headers, bytes)))
        }
        match value {
            OpOut::RawResponse(fut) => {
                let (status, headers, bytes) = fut.await;
                (status, ApiResponse::Success((headers, bytes)))
            }
            OpOut::Error(PocketIcError::CanisterRequestRoutingError(_)) => {
                from_bn_err(ErrorClientFacing::CanisterNotFound).await
            }
            OpOut::Error(PocketIcError::SubnetRequestRoutingError(_)) => {
                from_bn_err(ErrorClientFacing::SubnetNotFound).await
            }
            _ => (
                StatusCode::INTERNAL_SERVER_ERROR,
                ApiResponse::Error {
                    message: "operation returned invalid type".into(),
                },
            ),
        }
    }
}

// ----------------------------------------------------------------------------------------------------------------- //
// Read handlers

pub async fn handler_json_query(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    extract::Json(raw_canister_call): extract::Json<RawCanisterCall>,
) -> (StatusCode, Json<ApiResponse<RawCanisterResult>>) {
    let timeout = timeout_or_default(headers);
    match crate::pocket_ic::CanisterCall::try_from(raw_canister_call) {
        Ok(canister_call) => {
            let query_op = Query(canister_call);
            // TODO: how to know what run_operation returns, i.e. to what to parse it? (type safety?)
            // (applies to all handlers)
            let (code, response) = run_operation(api_state, instance_id, timeout, query_op).await;
            (code, Json(response))
        }
        Err(e) => (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::Error {
                message: format!("{e:?}"),
            }),
        ),
    }
}

pub async fn handler_topology(
    State(AppState { api_state, .. }): State<AppState>,
    headers: HeaderMap,
    Path(instance_id): Path<InstanceId>,
) -> (StatusCode, Json<ApiResponse<Topology>>) {
    let timeout = timeout_or_default(headers);
    let op = GetTopology {};
    let (code, response) = run_operation(api_state, instance_id, timeout, op).await;
    (code, Json(response))
}

pub async fn handler_get_time(
    State(AppState { api_state, .. }): State<AppState>,
    headers: HeaderMap,
    Path(instance_id): Path<InstanceId>,
) -> (StatusCode, Json<ApiResponse<RawTime>>) {
    let timeout = timeout_or_default(headers);
    let time_op = GetTime {};
    let (code, response) = run_operation(api_state, instance_id, timeout, time_op).await;
    (code, Json(response))
}

pub async fn handler_get_canister_http(
    State(AppState { api_state, .. }): State<AppState>,
    headers: HeaderMap,
    Path(instance_id): Path<InstanceId>,
) -> (StatusCode, Json<ApiResponse<Vec<RawCanisterHttpRequest>>>) {
    let timeout = timeout_or_default(headers);
    let op = GetCanisterHttp {};
    let (code, response) = run_operation(api_state, instance_id, timeout, op).await;
    (code, Json(response))
}

pub async fn handler_mock_canister_http(
    State(AppState { api_state, .. }): State<AppState>,
    headers: HeaderMap,
    Path(instance_id): Path<InstanceId>,
    axum::extract::Json(raw_mock_canister_http_response): axum::extract::Json<
        RawMockCanisterHttpResponse,
    >,
) -> (StatusCode, Json<ApiResponse<()>>) {
    let timeout = timeout_or_default(headers);
    let mock_canister_http_response: MockCanisterHttpResponse =
        raw_mock_canister_http_response.into();
    let op = MockCanisterHttp {
        mock_canister_http_response,
    };
    let (code, response) = run_operation(api_state, instance_id, timeout, op).await;
    (code, Json(response))
}

pub async fn handler_get_controllers(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    extract::Json(raw_canister_id): extract::Json<RawCanisterId>,
) -> (StatusCode, Json<ApiResponse<Vec<RawPrincipalId>>>) {
    let timeout = timeout_or_default(headers);
    match CanisterId::try_from(raw_canister_id.canister_id) {
        Ok(canister_id) => {
            let get_op = GetControllers { canister_id };
            let (code, response) = run_operation(api_state, instance_id, timeout, get_op).await;
            (code, Json(response))
        }
        Err(e) => (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::Error {
                message: format!("{e:?}"),
            }),
        ),
    }
}

pub async fn handler_get_cycles(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    extract::Json(raw_canister_id): extract::Json<RawCanisterId>,
) -> (StatusCode, Json<ApiResponse<RawCycles>>) {
    let timeout = timeout_or_default(headers);
    match CanisterId::try_from(raw_canister_id.canister_id) {
        Ok(canister_id) => {
            let get_op = GetCyclesBalance { canister_id };
            let (code, response) = run_operation(api_state, instance_id, timeout, get_op).await;
            (code, Json(response))
        }
        Err(e) => (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::Error {
                message: format!("{e:?}"),
            }),
        ),
    }
}

pub async fn handler_get_stable_memory(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    axum::extract::Json(raw_canister_id): axum::extract::Json<RawCanisterId>,
) -> (StatusCode, Json<ApiResponse<RawStableMemory>>) {
    let timeout = timeout_or_default(headers);
    match CanisterId::try_from(raw_canister_id.canister_id) {
        Ok(canister_id) => {
            let get_op = GetStableMemory { canister_id };
            let (code, response) = run_operation(api_state, instance_id, timeout, get_op).await;
            (code, Json(response))
        }
        Err(e) => (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::Error {
                message: format!("{e:?}"),
            }),
        ),
    }
}

pub async fn handler_get_subnet(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    axum::extract::Json(raw_canister_id): axum::extract::Json<RawCanisterId>,
) -> (StatusCode, Json<ApiResponse<Option<RawSubnetId>>>) {
    let timeout = timeout_or_default(headers);
    match CanisterId::try_from(raw_canister_id.canister_id) {
        Ok(canister_id) => {
            let op = GetSubnet { canister_id };
            let (code, res) = run_operation(api_state, instance_id, timeout, op).await;
            (code, Json(res))
        }
        Err(e) => (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::Error {
                message: format!("{e:?}"),
            }),
        ),
    }
}

pub async fn handler_pub_key(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    extract::Json(RawSubnetId { subnet_id }): extract::Json<RawSubnetId>,
) -> (StatusCode, Json<ApiResponse<Vec<u8>>>) {
    let timeout = timeout_or_default(headers);
    let subnet_id = ic_types::SubnetId::new(ic_types::PrincipalId(candid::Principal::from_slice(
        &subnet_id,
    )));
    let op = PubKey { subnet_id };
    let (code, res) = run_operation(api_state, instance_id, timeout, op).await;
    (code, Json(res))
}

pub async fn handler_canister_snapshot_download(
    State(AppState { api_state, .. }): State<AppState>,
    headers: HeaderMap,
    Path(instance_id): Path<InstanceId>,
    axum::extract::Json(RawCanisterSnapshotDownload {
        sender,
        canister_id,
        snapshot_id,
        snapshot_dir,
    }): axum::extract::Json<RawCanisterSnapshotDownload>,
) -> (StatusCode, Json<ApiResponse<()>>) {
    let timeout = timeout_or_default(headers);
    let canister_id = match CanisterId::try_from(canister_id.canister_id) {
        Ok(canister_id) => canister_id,
        Err(e) => {
            return (
                StatusCode::BAD_REQUEST,
                Json(ApiResponse::Error {
                    message: format!("{e:?}"),
                }),
            );
        }
    };
    let snapshot_id = match SnapshotId::try_from(snapshot_id) {
        Ok(snapshot_id) => snapshot_id,
        Err(e) => {
            return (
                StatusCode::BAD_REQUEST,
                Json(ApiResponse::Error {
                    message: format!("{e:?}"),
                }),
            );
        }
    };
    let op = CanisterSnapshotDownload {
        sender: ic_types::PrincipalId(sender.into()),
        canister_id,
        snapshot_id,
        snapshot_dir,
    };
    let (code, response) = run_operation(api_state, instance_id, timeout, op).await;
    (code, Json(response))
}

pub async fn handler_canister_snapshot_upload(
    State(AppState { api_state, .. }): State<AppState>,
    headers: HeaderMap,
    Path(instance_id): Path<InstanceId>,
    axum::extract::Json(RawCanisterSnapshotUpload {
        sender,
        canister_id,
        replace_snapshot,
        snapshot_dir,
    }): axum::extract::Json<RawCanisterSnapshotUpload>,
) -> (StatusCode, Json<ApiResponse<RawCanisterSnapshotId>>) {
    let timeout = timeout_or_default(headers);
    let canister_id = match CanisterId::try_from(canister_id.canister_id) {
        Ok(canister_id) => canister_id,
        Err(e) => {
            return (
                StatusCode::BAD_REQUEST,
                Json(ApiResponse::Error {
                    message: format!("{e:?}"),
                }),
            );
        }
    };
    let replace_snapshot = if let Some(replace_snapshot) = replace_snapshot {
        match SnapshotId::try_from(replace_snapshot.snapshot_id) {
            Ok(snapshot_id) => Some(snapshot_id),
            Err(e) => {
                return (
                    StatusCode::BAD_REQUEST,
                    Json(ApiResponse::Error {
                        message: format!("{e:?}"),
                    }),
                );
            }
        }
    } else {
        None
    };
    let op = CanisterSnapshotUpload {
        sender: ic_types::PrincipalId(sender.into()),
        canister_id,
        replace_snapshot,
        snapshot_dir,
    };
    let (code, response) = run_operation(api_state, instance_id, timeout, op).await;
    (code, Json(response))
}

pub async fn handler_dashboard(
    State(AppState { api_state, .. }): State<AppState>,
    NoApi(Path(instance_id)): NoApi<Path<InstanceId>>,
) -> (StatusCode, NoApi<Response<Body>>) {
    let op = DashboardRequest {};
    handle_raw(api_state, instance_id, op).await
}

pub async fn handler_status(
    State(AppState { api_state, .. }): State<AppState>,
    NoApi(Path(instance_id)): NoApi<Path<InstanceId>>,
    bytes: Bytes,
) -> (StatusCode, NoApi<Response<Body>>) {
    let op = StatusRequest { bytes };
    handle_raw(api_state, instance_id, op).await
}

async fn handler_call(
    State(AppState { api_state, .. }): State<AppState>,
    NoApi(Path((instance_id, effective_canister_id))): NoApi<Path<(InstanceId, CanisterId)>>,
    bytes: Bytes,
    version: CallRequestVersion,
) -> (StatusCode, NoApi<Response<Body>>) {
    let op = CallRequest {
        effective_canister_id,
        bytes,
        version,
    };
    handle_raw(api_state, instance_id, op).await
}

pub async fn handler_call_v2(
    state: State<AppState>,
    path: NoApi<Path<(InstanceId, CanisterId)>>,
    bytes: Bytes,
) -> (StatusCode, NoApi<Response<Body>>) {
    handler_call(state, path, bytes, CallRequestVersion::V2).await
}

pub async fn handler_call_v3(
    state: State<AppState>,
    path: NoApi<Path<(InstanceId, CanisterId)>>,
    bytes: Bytes,
) -> (StatusCode, NoApi<Response<Body>>) {
    handler_call(state, path, bytes, CallRequestVersion::V3).await
}

pub async fn handler_call_v4(
    state: State<AppState>,
    path: NoApi<Path<(InstanceId, CanisterId)>>,
    bytes: Bytes,
) -> (StatusCode, NoApi<Response<Body>>) {
    handler_call(state, path, bytes, CallRequestVersion::V4).await
}

async fn handler_query(
    State(AppState { api_state, .. }): State<AppState>,
    NoApi(Path((instance_id, effective_canister_id))): NoApi<Path<(InstanceId, CanisterId)>>,
    bytes: Bytes,
    version: query::Version,
) -> (StatusCode, NoApi<Response<Body>>) {
    let op = QueryRequest {
        effective_canister_id,
        bytes,
        version,
    };
    handle_raw(api_state, instance_id, op).await
}

pub async fn handler_query_v2(
    state: State<AppState>,
    path: NoApi<Path<(InstanceId, CanisterId)>>,
    bytes: Bytes,
) -> (StatusCode, NoApi<Response<Body>>) {
    handler_query(state, path, bytes, query::Version::V2).await
}

pub async fn handler_query_v3(
    state: State<AppState>,
    path: NoApi<Path<(InstanceId, CanisterId)>>,
    bytes: Bytes,
) -> (StatusCode, NoApi<Response<Body>>) {
    handler_query(state, path, bytes, query::Version::V3).await
}

async fn handler_canister_read_state(
    State(AppState { api_state, .. }): State<AppState>,
    NoApi(Path((instance_id, effective_canister_id))): NoApi<Path<(InstanceId, CanisterId)>>,
    bytes: Bytes,
    version: read_state::canister::Version,
) -> (StatusCode, NoApi<Response<Body>>) {
    let op = CanisterReadStateRequest {
        effective_canister_id,
        bytes,
        version,
    };
    handle_raw(api_state, instance_id, op).await
}

pub async fn handler_canister_read_state_v2(
    state: State<AppState>,
    path: NoApi<Path<(InstanceId, CanisterId)>>,
    bytes: Bytes,
) -> (StatusCode, NoApi<Response<Body>>) {
    handler_canister_read_state(state, path, bytes, read_state::canister::Version::V2).await
}

pub async fn handler_canister_read_state_v3(
    state: State<AppState>,
    path: NoApi<Path<(InstanceId, CanisterId)>>,
    bytes: Bytes,
) -> (StatusCode, NoApi<Response<Body>>) {
    handler_canister_read_state(state, path, bytes, read_state::canister::Version::V3).await
}

pub async fn handler_subnet_read_state(
    State(AppState { api_state, .. }): State<AppState>,
    NoApi(Path((instance_id, subnet_id))): NoApi<Path<(InstanceId, SubnetId)>>,
    bytes: Bytes,
    version: read_state::subnet::Version,
) -> (StatusCode, NoApi<Response<Body>>) {
    let op = SubnetReadStateRequest {
        subnet_id,
        bytes,
        version,
    };
    handle_raw(api_state, instance_id, op).await
}

pub async fn handler_subnet_read_state_v2(
    state: State<AppState>,
    path: NoApi<Path<(InstanceId, SubnetId)>>,
    bytes: Bytes,
) -> (StatusCode, NoApi<Response<Body>>) {
    handler_subnet_read_state(state, path, bytes, read_state::subnet::Version::V2).await
}

pub async fn handler_subnet_read_state_v3(
    state: State<AppState>,
    path: NoApi<Path<(InstanceId, SubnetId)>>,
    bytes: Bytes,
) -> (StatusCode, NoApi<Response<Body>>) {
    handler_subnet_read_state(state, path, bytes, read_state::subnet::Version::V3).await
}

async fn handle_raw<T: Operation + Send + Sync + 'static>(
    api_state: Arc<ApiState>,
    instance_id: InstanceId,
    op: T,
) -> (StatusCode, NoApi<Response<Body>>) {
    let (code, res): (StatusCode, ApiResponse<PocketHttpResponse>) =
        run_operation(api_state, instance_id, None, op).await;
    let response = match res {
        ApiResponse::Success((headers, bytes)) => {
            let mut resp = Response::builder().status(code);
            for (name, value) in headers {
                resp = resp.header(name, value);
            }
            resp.body(Body::from(bytes)).unwrap()
        }
        ApiResponse::Error { message } => make_plaintext_response(code, message),
        ApiResponse::Busy { .. } | ApiResponse::Started { .. } => {
            unreachable!("/api endpoints should always produce a response")
        }
    };
    (code, NoApi(response))
}

/// The result of a long running PocketIC operation is stored in a graph as an OpOut variant.
/// When polling, the type (and therefore the variant) is no longer known. Therefore we need
/// to try every variant and immediately convert to an axum::Response so that axum understands
/// the return type.
async fn op_out_to_response(op_out: OpOut) -> Response {
    match op_out {
        opout @ OpOut::MessageId(_) => (
            StatusCode::OK,
            Json(ApiResponse::Success(
                Result::<RawMessageId, RejectResponse>::try_from(opout).unwrap(),
            )),
        )
            .into_response(),
        OpOut::NoOutput => (
            StatusCode::OK,
            Json(ApiResponse::Success(())).into_response(),
        )
            .into_response(),
        OpOut::Topology(topology) => {
            (StatusCode::OK, Json(ApiResponse::Success(topology))).into_response()
        }
        opout @ OpOut::Time(_) => (
            StatusCode::OK,
            Json(ApiResponse::Success(RawTime::try_from(opout).unwrap())),
        )
            .into_response(),
        opout @ OpOut::CanisterResult(_) => (
            StatusCode::OK,
            Json(ApiResponse::Success(
                RawCanisterResult::try_from(opout).unwrap(),
            )),
        )
            .into_response(),
        opout @ OpOut::CanisterId(_) => (
            StatusCode::OK,
            Json(ApiResponse::Success(
                RawCanisterId::try_from(opout).unwrap(),
            )),
        )
            .into_response(),
        opout @ OpOut::Controllers(_) => (
            StatusCode::OK,
            Json(ApiResponse::Success(
                Vec::<RawPrincipalId>::try_from(opout).unwrap(),
            )),
        )
            .into_response(),
        opout @ OpOut::Cycles(_) => (
            StatusCode::OK,
            Json(ApiResponse::Success(RawCycles::try_from(opout).unwrap())),
        )
            .into_response(),
        opout @ OpOut::Bytes(_) => (
            StatusCode::OK,
            Json(ApiResponse::Success(Vec::<u8>::try_from(opout).unwrap())),
        )
            .into_response(),
        opout @ OpOut::StableMemBytes(_) => (
            StatusCode::OK,
            Json(ApiResponse::Success(
                RawStableMemory::try_from(opout).unwrap(),
            )),
        )
            .into_response(),
        opout @ OpOut::MaybeSubnetId(_) => (
            StatusCode::OK,
            Json(ApiResponse::Success(
                Option::<RawSubnetId>::try_from(opout).unwrap(),
            )),
        )
            .into_response(),
        OpOut::Error(PocketIcError::Forbidden(msg)) => (
            StatusCode::FORBIDDEN,
            Json(ApiResponse::<()>::Error {
                message: msg.to_string(),
            }),
        )
            .into_response(),
        opout @ OpOut::Error(_) => (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::<()>::Error {
                message: format!("{:?}", PocketIcError::try_from(opout).unwrap()),
            }),
        )
            .into_response(),
        opout @ OpOut::CanisterHttp(_) => (
            StatusCode::OK,
            Json(ApiResponse::Success(
                Vec::<RawCanisterHttpRequest>::try_from(opout).unwrap(),
            )),
        )
            .into_response(),
        opout @ OpOut::CanisterSnapshotId(_) => (
            StatusCode::OK,
            Json(ApiResponse::Success(
                RawCanisterSnapshotId::try_from(opout).unwrap(),
            )),
        )
            .into_response(),
        OpOut::RawResponse(fut) => {
            let (status, headers, bytes) = fut.await;
            let mut resp = Response::builder().status(status);
            for (name, value) in headers {
                resp = resp.header(name, value);
            }
            resp.body(Body::from(bytes)).unwrap()
        }
    }
}

/// Read a result. Needed for polling for a previous ApiResponse::Started reply.
pub async fn handler_read_graph(
    State(AppState { api_state, .. }): State<AppState>,
    // TODO: type state label and op id correctly but such that axum can handle it
    Path((state_label_str, op_id_str)): Path<(String, String)>,
) -> Response {
    let Ok(vec) = base64::decode_config(state_label_str.as_bytes(), base64::URL_SAFE) else {
        return (StatusCode::BAD_REQUEST, "Malformed state label.").into_response();
    };
    if let Ok(state_label) = StateLabel::try_from(vec) {
        let op_id = OpId(op_id_str.clone());
        // TODO: use new_state_label and return it to library
        match api_state.read_graph(state_label, op_id).await {
            Some((_new_state_label, op_out)) => op_out_to_response(op_out).await,
            _ => (
                StatusCode::NOT_FOUND,
                Json(ApiResponse::<()>::Error {
                    message: format!(
                        "state_label / op_id not found: {} / {}",
                        state_label_str, op_id_str,
                    ),
                }),
            )
                .into_response(),
        }
    } else {
        (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::<()>::Error {
                message: "Malformed state label.".to_string(),
            }),
        )
            .into_response()
    }
}

/// Prune a result. Needed for releasing memory after successfully polling for a previous ApiResponse::Started reply.
pub async fn handler_prune_graph(
    State(AppState { api_state, .. }): State<AppState>,
    Path((state_label_str, op_id_str)): Path<(String, String)>,
) -> (StatusCode, Json<ApiResponse<()>>) {
    let Ok(vec) = base64::decode_config(state_label_str.as_bytes(), base64::URL_SAFE) else {
        return (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::<()>::Error {
                message: "Malformed state label.".to_string(),
            }),
        );
    };
    if let Ok(state_label) = StateLabel::try_from(vec) {
        let op_id = OpId(op_id_str.clone());
        match api_state.prune_graph(state_label, op_id).await {
            Some(()) => (StatusCode::OK, Json(ApiResponse::Success(()))),
            None => (
                StatusCode::NOT_FOUND,
                Json(ApiResponse::<()>::Error {
                    message: format!(
                        "state_label / op_id not found: {} / {}",
                        state_label_str, op_id_str,
                    ),
                }),
            ),
        }
    } else {
        (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::<()>::Error {
                message: "Malformed state label.".to_string(),
            }),
        )
    }
}

// ----------------------------------------------------------------------------------------------------------------- //
// Update handlers

pub async fn handler_submit_ingress_message(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    extract::Json(raw_canister_call): extract::Json<RawCanisterCall>,
) -> (
    StatusCode,
    Json<ApiResponse<Result<RawMessageId, RejectResponse>>>,
) {
    let timeout = timeout_or_default(headers);
    match crate::pocket_ic::CanisterCall::try_from(raw_canister_call) {
        Ok(canister_call) => {
            let ingress_op = SubmitIngressMessage(canister_call);
            let (code, response) = run_operation(api_state, instance_id, timeout, ingress_op).await;
            (code, Json(response))
        }
        Err(e) => (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::Error {
                message: format!("{e:?}"),
            }),
        ),
    }
}

pub async fn handler_await_ingress_message(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    extract::Json(raw_message_id): extract::Json<RawMessageId>,
) -> (StatusCode, Json<ApiResponse<RawCanisterResult>>) {
    let timeout = timeout_or_default(headers);
    match crate::pocket_ic::MessageId::try_from(raw_message_id) {
        Ok(message_id) => {
            let ingress_op = AwaitIngressMessage(message_id);
            let (code, response) = run_operation(api_state, instance_id, timeout, ingress_op).await;
            (code, Json(response))
        }
        Err(e) => (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::Error {
                message: format!("{e:?}"),
            }),
        ),
    }
}

pub async fn handler_ingress_status(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    extract::Json(raw_ingress_status_args): extract::Json<RawIngressStatusArgs>,
) -> (StatusCode, Json<ApiResponse<Option<RawCanisterResult>>>) {
    let timeout = timeout_or_default(headers);
    match crate::pocket_ic::MessageId::try_from(raw_ingress_status_args.raw_message_id) {
        Ok(message_id) => {
            let ingress_op = IngressMessageStatus {
                message_id,
                caller: raw_ingress_status_args
                    .raw_caller
                    .map(|caller| caller.into()),
            };
            let (code, response) = run_operation(api_state, instance_id, timeout, ingress_op).await;
            (code, Json(response))
        }
        Err(e) => (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::Error {
                message: format!("{e:?}"),
            }),
        ),
    }
}

pub async fn handler_set_time(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    axum::extract::Json(time): axum::extract::Json<rest::RawTime>,
) -> (StatusCode, Json<ApiResponse<()>>) {
    let timeout = timeout_or_default(headers);
    let op = SetTime {
        time: ic_types::Time::from_nanos_since_unix_epoch(time.nanos_since_epoch),
    };
    let (code, response) = run_operation(api_state, instance_id, timeout, op).await;
    (code, Json(response))
}

pub async fn handler_set_certified_time(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    axum::extract::Json(time): axum::extract::Json<rest::RawTime>,
) -> (StatusCode, Json<ApiResponse<()>>) {
    let timeout = timeout_or_default(headers);
    let op = SetCertifiedTime {
        time: ic_types::Time::from_nanos_since_unix_epoch(time.nanos_since_epoch),
    };
    let (code, response) = run_operation(api_state, instance_id, timeout, op).await;
    (code, Json(response))
}

pub async fn handler_add_cycles(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    extract::Json(raw_add_cycles): extract::Json<RawAddCycles>,
) -> (StatusCode, Json<ApiResponse<RawCycles>>) {
    let timeout = timeout_or_default(headers);
    match AddCycles::try_from(raw_add_cycles) {
        Ok(add_op) => {
            let (code, response) = run_operation(api_state, instance_id, timeout, add_op).await;
            (code, Json(response))
        }
        Err(e) => (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::Error {
                message: format!("{e:?}"),
            }),
        ),
    }
}

pub async fn handler_set_stable_memory(
    State(AppState {
        api_state,
        blob_store,
        ..
    }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    axum::extract::Json(raw): axum::extract::Json<RawSetStableMemory>,
) -> (StatusCode, Json<ApiResponse<()>>) {
    let timeout = timeout_or_default(headers);
    match SetStableMemory::from_store(raw, blob_store).await {
        Ok(set_op) => {
            let (code, response) = run_operation(api_state, instance_id, timeout, set_op).await;
            (code, Json(response))
        }
        Err(e) => (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::Error {
                message: format!("{e:?}"),
            }),
        ),
    }
}

pub async fn handler_tick(
    State(AppState { api_state, .. }): State<AppState>,
    Path(instance_id): Path<InstanceId>,
    headers: HeaderMap,
    axum::extract::Json(ticks_configs): axum::extract::Json<TickConfigs>,
) -> (StatusCode, Json<ApiResponse<()>>) {
    let timeout = timeout_or_default(headers);
    let op = Tick {
        configs: ticks_configs,
    };
    let (code, res) = run_operation(api_state, instance_id, timeout, op).await;
    (code, Json(res))
}

// ----------------------------------------------------------------------------------------------------------------- //
// Other handlers

pub async fn status() -> StatusCode {
    StatusCode::OK
}

fn contains_unimplemented(config: ExtendedSubnetConfigSet) -> bool {
    Iterator::any(
        &mut vec![
            config.nns,
            config.sns,
            config.ii,
            config.fiduciary,
            config.bitcoin,
        ]
        .into_iter()
        .flatten()
        .chain(config.system)
        .chain(config.application)
        .chain(config.verified_application),
        |spec| !spec.is_supported(),
    )
}

/// Create a new empty IC instance from a given subnet configuration.
/// The new InstanceId will be returned.
pub async fn create_instance(
    State(AppState {
        api_state,
        runtime,
        mainnet_routing_table,
        ..
    }): State<AppState>,
    extract::Json(instance_config): extract::Json<InstanceConfig>,
) -> (StatusCode, Json<rest::CreateInstanceResponse>) {
    let mut subnet_configs = instance_config.subnet_config_set;

    let skip_validate_subnet_configs = instance_config
        .state_dir
        .as_ref()
        .map(|state_dir| File::open(state_dir.clone().join("topology.json")).is_ok())
        .unwrap_or_default();
    if !skip_validate_subnet_configs {
        if let Some(ref icp_features) = instance_config.icp_features {
            subnet_configs = match subnet_configs.try_with_icp_features(icp_features) {
                Ok(subnet_configs) => subnet_configs,
                Err(e) => {
                    return (
                        StatusCode::BAD_REQUEST,
                        Json(rest::CreateInstanceResponse::Error {
                            message: format!("Subnet config failed to validate: {e}"),
                        }),
                    );
                }
            };
        }
        if let Err(e) = subnet_configs.validate() {
            return (
                StatusCode::BAD_REQUEST,
                Json(rest::CreateInstanceResponse::Error {
                    message: format!("Subnet config failed to validate: {e:?}"),
                }),
            );
        }
    }
    // TODO: Remove this once the SubnetStateConfig variants are implemented
    if contains_unimplemented(subnet_configs.clone()) {
        return (
            StatusCode::BAD_REQUEST,
            Json(rest::CreateInstanceResponse::Error {
                message: "SubnetStateConfig::FromBlobStore is not yet implemented".to_owned(),
            }),
        );
    }

    let log_level = if let Some(log_level) = instance_config.log_level {
        match Level::from_str(&log_level) {
            Ok(log_level) => Some(log_level),
            Err(e) => {
                return (
                    StatusCode::BAD_REQUEST,
                    Json(rest::CreateInstanceResponse::Error {
                        message: format!("Failed to parse log level: {e:?}"),
                    }),
                );
            }
        }
    } else {
        None
    };

    let initial_time = match instance_config.initial_time {
        Some(InitialTime::Timestamp(raw_time)) => Some(
            ic_types::Time::from_nanos_since_unix_epoch(raw_time.nanos_since_epoch),
        ),
        Some(InitialTime::AutoProgress(_)) => Some(SystemTime::now().try_into().unwrap()),
        None => None,
    };
    let auto_progress = match instance_config.initial_time {
        Some(InitialTime::AutoProgress(config)) => Some(config),
        Some(InitialTime::Timestamp(_)) | None => None,
    };
    let auto_progress_enabled = auto_progress.is_some();

    if let Some(ref icp_features) = instance_config.icp_features {
        // using `let IcpFeatures { }` with explicit field names
        // to force an update after adding a new field to `IcpFeatures`
        let IcpFeatures {
            /* `nns_ui` does not depend on `registry` */
            registry: _,
            cycles_minting,
            icp_token,
            /* `nns_ui` does not depend on `cycles_token` */
            cycles_token: _,
            nns_governance,
            sns,
            ii,
            nns_ui,
            /* `nns_ui` does not depend on `bitcoin` */
            bitcoin: _,
            /* `nns_ui` does not depend on `dogecoin` */
            dogecoin: _,
            /* `nns_ui` does not depend on `canister_migration` */
            canister_migration: _,
        } = icp_features;
        if nns_ui.is_some() {
            if instance_config.http_gateway_config.is_none() {
                return (
                    StatusCode::BAD_REQUEST,
                    Json(rest::CreateInstanceResponse::Error {
                        message: "The `nns_ui` feature requires an HTTP gateway to be created via `http_gateway_config`.".to_string()
                    }),
                );
            }
            for (icp_feature, icp_feature_str) in [
                (cycles_minting, "cycles_minting"),
                (icp_token, "icp_token"),
                (nns_governance, "nns_governance"),
                (sns, "sns"),
                (ii, "ii"),
            ] {
                if icp_feature.is_none() {
                    return (
                        StatusCode::BAD_REQUEST,
                        Json(rest::CreateInstanceResponse::Error {
                            message: format!(
                                "The `nns_ui` feature requires the `{icp_feature_str}` feature to be enabled, too."
                            ),
                        }),
                    );
                }
            }
        }
    }

    match api_state
        .add_instance(
            move |seed, gateway_port| {
                PocketIc::try_new(
                    runtime,
                    mainnet_routing_table,
                    seed,
                    subnet_configs,
                    instance_config.state_dir,
                    instance_config.icp_config.unwrap_or(IcpConfig::default()),
                    log_level,
                    instance_config.bitcoind_addr,
                    instance_config.dogecoind_addr,
                    instance_config.icp_features,
                    instance_config.incomplete_state,
                    initial_time,
                    auto_progress_enabled,
                    gateway_port,
                )
            },
            auto_progress,
            instance_config.http_gateway_config,
        )
        .await
    {
        Ok((instance_id, topology, http_gateway_info)) => (
            StatusCode::CREATED,
            Json(rest::CreateInstanceResponse::Created {
                instance_id,
                topology,
                http_gateway_info,
            }),
        ),
        Err(err) => (
            StatusCode::BAD_REQUEST,
            Json(rest::CreateInstanceResponse::Error { message: err }),
        ),
    }
}

pub async fn list_instances(
    State(AppState { api_state, .. }): State<AppState>,
) -> Json<Vec<String>> {
    let instances = api_state.list_instance_states().await;
    Json(instances)
}

pub async fn delete_instance(
    State(AppState { api_state, .. }): State<AppState>,
    Path(id): Path<InstanceId>,
) -> StatusCode {
    api_state.delete_instance(id).await;
    StatusCode::OK
}

pub async fn list_http_gateways(
    State(AppState { api_state, .. }): State<AppState>,
) -> Json<Vec<HttpGatewayDetails>> {
    let http_gateways = api_state.list_http_gateways().await;
    Json(http_gateways)
}

/// Create a new HTTP gateway instance from a given HTTP gateway configuration.
/// The new InstanceId and HTTP gateway's port will be returned.
pub async fn create_http_gateway(
    State(AppState { api_state, .. }): State<AppState>,
    extract::Json(http_gateway_config): extract::Json<HttpGatewayConfig>,
) -> (StatusCode, Json<rest::CreateHttpGatewayResponse>) {
    let listener = match api_state.create_http_gateway_listener(
        http_gateway_config.ip_addr.clone(),
        http_gateway_config.port,
    ) {
        Ok(listener) => listener,
        Err(e) => {
            return (
                StatusCode::BAD_GATEWAY,
                Json(rest::CreateHttpGatewayResponse::Error { message: e }),
            );
        }
    };
    match api_state
        .create_http_gateway(http_gateway_config, listener)
        .await
    {
        Ok(http_gateway_info) => (
            StatusCode::CREATED,
            Json(rest::CreateHttpGatewayResponse::Created(http_gateway_info)),
        ),
        Err(e) => (
            StatusCode::BAD_GATEWAY,
            Json(rest::CreateHttpGatewayResponse::Error { message: e }),
        ),
    }
}

/// Stops an HTTP gateway instance.
pub async fn stop_http_gateway(
    State(AppState { api_state, .. }): State<AppState>,
    Path(id): Path<InstanceId>,
) -> (StatusCode, Json<ApiResponse<()>>) {
    api_state.stop_http_gateway(id).await;
    (StatusCode::OK, Json(ApiResponse::Success(())))
}

pub async fn auto_progress(
    State(AppState { api_state, .. }): State<AppState>,
    Path(id): Path<InstanceId>,
    extract::Json(auto_progress_config): extract::Json<AutoProgressConfig>,
) -> (StatusCode, Json<ApiResponse<()>>) {
    if let Err(e) = api_state
        .auto_progress(id, auto_progress_config.artificial_delay_ms)
        .await
    {
        (
            StatusCode::BAD_REQUEST,
            Json(ApiResponse::Error { message: e }),
        )
    } else {
        (StatusCode::OK, Json(ApiResponse::Success(())))
    }
}

pub async fn get_auto_progress(
    State(AppState { api_state, .. }): State<AppState>,
    Path(id): Path<InstanceId>,
) -> (StatusCode, Json<ApiResponse<bool>>) {
    let auto_progress = api_state.get_auto_progress(id).await;
    (StatusCode::OK, Json(ApiResponse::Success(auto_progress)))
}

pub async fn stop_progress(
    State(AppState { api_state, .. }): State<AppState>,
    Path(id): Path<InstanceId>,
) -> (StatusCode, Json<ApiResponse<()>>) {
    api_state.stop_progress(id).await;
    (StatusCode::OK, Json(ApiResponse::Success(())))
}

pub trait RouterExt<S>
where
    S: Clone + Send + Sync + 'static,
{
    fn directory_route(self, path: &str, method_router: ApiMethodRouter<S>) -> Self;
}

impl<S> RouterExt<S> for ApiRouter<S>
where
    S: Clone + Send + Sync + 'static,
{
    fn directory_route(self, path: &str, method_router: ApiMethodRouter<S>) -> Self {
        self.api_route(path, method_router.clone())
            .route(&format!("{path}/"), method_router)
    }
}

/// A typed header that a client can use to specify the maximum duration it is willing to wait for a
/// synchronous response.
pub struct ProcessingTimeout(pub Duration);

impl headers::Header for ProcessingTimeout {
    fn name() -> &'static http::header::HeaderName {
        &TIMEOUT_HEADER_NAME
    }

    fn decode<'i, I>(values: &mut I) -> Result<Self, headers::Error>
    where
        I: Iterator<Item = &'i http::header::HeaderValue>,
    {
        fn to_invalid<E>(_: E) -> headers::Error {
            headers::Error::invalid()
        }

        let value = values.next().ok_or_else(headers::Error::invalid)?;
        let nanos = value
            .to_str()
            .map_err(to_invalid)?
            .parse::<u64>()
            .map_err(to_invalid)?;
        Ok(Self(Duration::from_millis(nanos)))
    }

    fn encode<E>(&self, values: &mut E)
    where
        E: Extend<http::header::HeaderValue>,
    {
        let nanos = self.0.as_millis();
        let value = http::header::HeaderValue::from_str(&format!("{nanos}")).unwrap();

        values.extend(std::iter::once(value));
    }
}

pub fn timeout_or_default(header_map: HeaderMap) -> Option<Duration> {
    let timeout_from_header = header_map.typed_get::<ProcessingTimeout>().map(|x| x.0);
    Some(timeout_from_header.unwrap_or(DEFAULT_SYNC_WAIT_DURATION))
}
